package com.jy.rhin.facade.infra.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
import com.jy.rhin.application.infra.NotificationApplicationService;
import com.jy.rhin.domain.exception.ListenerException;
import com.jy.rhin.infra.repo.notfication.message.aspect.Receiver;
import com.jy.rhin.sdk.dto.notification.NotificationMessageDto;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.Objects;

/**
 * @author dhl
 * @date 2022/7/5
 * @Description
 */
@Slf4j
@Component
public class MessageListener {

    @Resource
    private NotificationApplicationService notificationApplicationService;

    private static final String TOPIC_EXP = "topicExp";

    private static final String TOPIC_EXP_SPLIT = "rhin:personIdentifierSplit";

    @RabbitListener(queues = "${rhin.amqp.notificationQueue}")
    @Receiver
    public void handleDelivery(Message message, Channel channel) {
        String body = new String(message.getBody());
        try {
            NotificationMessageDto content = checkMessage(body);
            String data = JSON.toJSONString(content.getData());
            if (needMessage(data)) {
                notificationApplicationService.receive(data, content.getMsgId(), content.getSender(), new Date());
            }
        } catch (ListenerException e) {
            log.error("监听消息异常，不符合消息格式:{}-{}", body, e.getMessage());
            throw e;
        } catch (Exception e) {
            log.error("{}消息处理失败", body, e);
            throw e;
        } finally {
            try {
                //消息消费成功或者失败都ack确认
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            } catch (IOException e) {
                log.error("确认消息失败 ", e);
            }
        }
    }

    /**
     * 校验消息格式
     *
     * @param body
     */
    private NotificationMessageDto checkMessage(String body) {
        try {
            log.debug("收到业务服务发来的消息：{}", body);
            NotificationMessageDto content = NotificationMessageDto.from(body);
            if (ObjectUtils.isNull(content.getData()) || StringUtils.isEmpty(content.getMsgId())) {
                log.error("消息内容缺失");
                throw new ListenerException(ListenerException.Error.LISTENER_MSG_CONTENT_NULL);
            }
            return content;
        } catch (ListenerException e) {
            throw e;
        } catch (Exception e) {
            log.error("消息格式化错误");
            throw new ListenerException(ListenerException.Error.LISTENER_MSG_FORMAT_ERROR);
        }
    }

    /**
     * 判断是否是需要处理的消息
     *
     * @param data
     * @return
     */
    private boolean needMessage(String data) {
        JSONObject messageContent = JSON.parseObject(data, Feature.IgnoreAutoType);
        String topicExp = messageContent.getString(TOPIC_EXP);
        if (StringUtils.isEmpty(topicExp)) {
            log.error("消息主题表达式缺失");
            throw new ListenerException(ListenerException.Error.LISTENER_RECEIVE_TOPIC_NOT_FOUND);
        }
        if (Objects.equals(TOPIC_EXP_SPLIT, topicExp)) {
            log.debug("居民主索引拆分消息,不做处理");
            return false;
        }
        return true;
    }
}
